/*
 * Copyright 2008-2009 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package net.hasor.rsf.rpc.caller.remote;

import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import net.hasor.rsf.InterAddress;
import net.hasor.rsf.RsfContext;
import net.hasor.rsf.RsfEnvironment;
import net.hasor.rsf.RsfSettings;
import net.hasor.rsf.container.RsfBeanContainer;
import net.hasor.rsf.domain.ProtocolStatus;
import net.hasor.rsf.domain.RequestInfo;
import net.hasor.rsf.domain.ResponseInfo;
import net.hasor.rsf.rpc.caller.RsfCaller;
import net.hasor.rsf.utils.ExecutesManager;
import net.hasor.rsf.utils.ProtocolUtils;

/**
 * 扩展{@link RsfCaller}，用来支持远程机器发来的调用请求。
 *
 * @author 赵永春(zyc@hasor.net)
 * @version : 2015年12月8日
 */

// TODO: 17/5/6 by zmyer
public class RemoteRsfCaller extends RsfCaller {
    //执行器管理器
    private final ExecutesManager executesManager;
    //发送者监听器
    private final RemoteSenderListener senderListener;

    // TODO: 17/5/6 by zmyer
    public RemoteRsfCaller(RsfContext rsfContext, RsfBeanContainer rsfBeanContainer,
        RemoteSenderListener senderListener) {
        super(rsfContext, rsfBeanContainer, senderListener);
        //设置发送者监听器
        this.senderListener = senderListener;
        //从上下文中读取对应的配置对象
        RsfSettings rsfSettings = rsfContext.getSettings();
        //读取队列最大长度
        int queueSize = rsfSettings.getQueueMaxSize();
        //线程池中线程最小数目
        int minCorePoolSize = rsfSettings.getQueueMinPoolSize();
        //线程池中线程最大数目
        int maxCorePoolSize = rsfSettings.getQueueMaxPoolSize();
        //连接存活时间
        long keepAliveTime = rsfSettings.getQueueKeepAliveTime();
        //创建执行器管理器
        this.executesManager = new ExecutesManager(minCorePoolSize, maxCorePoolSize, queueSize, keepAliveTime, rsfContext.getClassLoader());
    }

    // TODO: 17/5/6 by zmyer
    /** 销毁。 */
    public void shutdown() {
        logger.info("rsfCaller -> shutdown.");
        //关闭执行器
        this.executesManager.shutdown();
    }

    /**
     * 收到Request请求，并将该请求安排进队列，由队列安排方法调用。
     *
     * @param target 目标调用地址。
     * @param info 请求消息。
     */
    // TODO: 17/5/6 by zmyer
    public void onRequest(InterAddress target, RequestInfo info) {
        //读取环境变量
        RsfEnvironment rsfEnv = this.getContext().getEnvironment();
        //构建服务唯一名称
        String serviceUniqueName = "[" + info.getServiceGroup() + "]" + info.getServiceName() + "-" + info.getServiceVersion();
        try {
            invLogger.info("request({}) -> received, bindID ={}, targetMethod ={}, remoteAddress ={}.", //
                info.getRequestID(), serviceUniqueName, info.getTargetMethod(), target);

            //根据服务唯一名称,从执行器管理对象中读取对应的执行器
            Executor executor = executesManager.getExecute(serviceUniqueName);
            //放入业务线程准备执行
            executor.execute(new RemoteRsfCallerProcessing(target, this, info));
            //创建请求应答对象
            ResponseInfo resp = ProtocolUtils.buildResponseStatus(rsfEnv, info.getRequestID(),
                ProtocolStatus.Accept, null);
            //开始发送应答消息
            this.senderListener.sendResponse(target, resp);
        } catch (RejectedExecutionException e) {
            invLogger.info("request({}) -> rejected request, queue is full. -> bindID ={}, targetMethod ={}, remoteAddress ={}.", //
                info.getRequestID(), serviceUniqueName, info.getTargetMethod(), target);
            //
            String errorMessage = "(" + e.getClass().getName() + ")" + e.getMessage();
            String msgLog = "rejected request, queue is full." + errorMessage;
            logger.warn(msgLog, e);
            ResponseInfo resp = ProtocolUtils.buildResponseStatus(rsfEnv, info.getRequestID(), ProtocolStatus.QueueFull, msgLog);
            this.senderListener.sendResponse(target, resp);
        }
    }
    //

    // TODO: 17/5/6 by zmyer
    /** 获取消息监听器。 */
    RemoteSenderListener getSenderListener() {
        return this.senderListener;
    }
}